import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Demo09_cdc {

    public static void cdc_dataStream() throws Exception{
        //1.获取Flink执行环境
        StreamExecutionEnvironment environment =
                StreamExecutionEnvironment.getExecutionEnvironment();
        environment.setParallelism(1);
        //2.使用FlinkCDC构造MySQLSource
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("localhost")
                .port(3306)
                .username("root")
                .password("root")
                //监控的数据库列表，可以同时监控多个数据库
                .databaseList("cdc,cdc_other")
                //这里最好写上数据库名，明确监控的是哪个数据库的表、
                //如不写数据库名，则表示监控所有databaseList数据库中此名字的表
                .tableList("cdc.student1","cdc.student2","cdc.student3")
                //设置从哪里开始读取数据
                .startupOptions(StartupOptions.latest())
                .deserializer(new JsonDebeziumDeserializationSchema())
                .build();
        //3.读取数据
        DataStreamSource<String> mysqlDS =
                environment.fromSource(
                        mySqlSource,
                        WatermarkStrategy.noWatermarks(),
                        "mysql-source"
                );
        //4.打印
        mysqlDS.print();
        //5.启动
        environment.execute();
    }

    public static void cdc_sql(){
        //1.获取Flink执行环境
        StreamExecutionEnvironment environment =
                StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(environment);
        //2.使用 FlinkCDC SQL 方式建表
        tableEnvironment.executeSql(
                "CREATE TABLE mysql_binlog ( " +
                        " id INT primary key NOT ENFORCED," +
                        " name STRING" +
                        ") WITH (" +
                        " 'connector' = 'mysql-cdc'," +
                        " 'hostname' = 'localhost'," +
                        " 'port' = '3306'," +
                        " 'username' = 'root'," +
                        " 'password' = 'root'," +
                        " 'database-name' = 'cdc'," +
                        " 'table-name' = 'student1'" +
                        ")"
        );

        //3.查询表中数据
        TableResult tableResult = tableEnvironment.executeSql("select * from mysql_binlog");
        //4.打印结果
        tableResult.print();
    }

    public static void main(String[] args) throws Exception {

        cdc_sql();

    }
}
